package cn.lsh.kafka.producer;

import cn.lsh.kafka.db.entity.OffsetEntity;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.time.LocalTime;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class OffsetProducer {

	public static void main(String[] args) {
		Properties properties = new Properties();
		//配置broker地址
		properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		//key序列化类
		properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
		//value序列化类
		properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "cn.lsh.kafka.serializer.OffsetSerializer");
		properties.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "cn.lsh.kafka.interceptor.TimeProducerInterceptor,cn.lsh.kafka.interceptor.CountProducerInterceptor");
		properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "cn.lsh.kafka.partitioner.MyProducerPartitioner");



		KafkaProducer<Integer, OffsetEntity> producer = new KafkaProducer<>(properties);
		for (int i = 0; i < 10; i++) {
			//构建一个ProducerRecord，需要指定topic，可以指定分区号，
			OffsetEntity entity = new OffsetEntity();
			entity.setValue(100L);
			entity.setGroupId("group_1");
			entity.setTopic("ddd-1");
			entity.setCreateDate(new Date());
			ProducerRecord<Integer, OffsetEntity> record = new ProducerRecord<>("ddd-1", i,entity);
			Future<RecordMetadata> send = producer.send(record, (recordMetadata, e) -> {
				if (e == null) {
					System.out.println("消息发送成功：偏移量=" + recordMetadata.offset() + "，分区=" + recordMetadata.partition());
				} else {
					System.out.println("消息发送失败：" + e);
				}
			});

			try {
				//get方法会阻塞当前线程，即同步发送
				RecordMetadata recordMetadata = send.get();
				System.out.println("--------消息发送成功：偏移量=" + recordMetadata.offset() + "，分区=" + recordMetadata.partition());

			} catch (ExecutionException | InterruptedException e) {
				e.printStackTrace();
			}
		}
		System.out.println("消息已生产完：" + LocalTime.now());
		// Thread.sleep(10000);

		//close关闭资源，会关闭RecordAccumulator缓冲区，即close会把剩余的消息发送出去；并且拦截器以及分区器的close方法也是由这个close触发调用
		producer.close();
	}
}
